package stomp

import (
	"container/list"
	"fmt"
	"net"
	"sync"

	"github.com/fwhezfwhez/errorx"
	"github.com/go-stomp/stomp/v3"
)

//支持stom 协议的1.0, 1.1,1.2这三个版本

// MsgQueue 消息队列的相关操作
type MsgQueue struct {
	connection *stomp.Conn
	subs       *list.List
	host       string
	port       string
	user       string
	password   string
	closed     bool

	mutex sync.Mutex

	DisconnectCallback func()
}

const (
	//Topic 主题
	Topic string = "/topic/"
	//Queue 队列
	Queue string = "/queue/"
)

// ConnectToServer 连接到服务器,并开始接收来自队列的消息
// 参数:
//
//	host	:主机
//	port	:端口
//	user	:用户
//	pwd		:密码
//	queues	:需要接收消息的队列
//	callback:接收到消息的回调函数
//			:回调函数参数说明
//			msg		 :消息
//			queueName:收到消息的队列名称
//
// 返回：
//
//	error	:nil或者错误信息
func (msgQueue *MsgQueue) ConnectToServer(host, port, user, pwd string, queues []string, callback func(msg, queueName string)) error {
	msgQueue.closed = true
	conn, err := msgQueue.connect(host, port, user, pwd)
	if err != nil {
		return errorx.New(err)
	}

	msgQueue.connection = conn

	if msgQueue.subs == nil {
		msgQueue.subs = list.New()
	}

	if queues != nil || len(queues) > 0 {
		err = msgQueue.SubscribeQueue(queues, callback)
		return fmt.Errorf("订阅队列发生异常,错误:%v", err)
	}
	msgQueue.closed = false
	return nil
}

// Disconnect 断开连接
func (msgQueue *MsgQueue) Disconnect() {
	msgQueue.closed = true
	if msgQueue.subs != nil {
		var next *list.Element
		for sub := msgQueue.subs.Front(); sub != nil; sub = next {
			o, r := sub.Value.(*stomp.Subscription)
			if r {
				_ = o.Unsubscribe()
			}
			next = sub.Next()
			msgQueue.subs.Remove(sub)
		}
	}
	if msgQueue.connection != nil {
		_ = msgQueue.connection.Disconnect()
	}
}

// Connect 连接到消息服务器,使用tcp的方式连接
// 参数：
//
//	host	:主机地址
//	port	:连接端口
//	user	:用户名
//	password:密码
//
// 返回：
//
//	*stomp.Conn:连接成功后的连对象
//		  error:错误信息
func (msgQueue *MsgQueue) connect(host, port, user, password string) (*stomp.Conn, error) {
	var options []func(*stomp.Conn) error = []func(*stomp.Conn) error{
		stomp.ConnOpt.HeartBeat(10, 10),
		stomp.ConnOpt.Login(user, password),
	}
	conn, err := stomp.Dial("tcp", net.JoinHostPort(host, port), options...)
	if nil == err {
		msgQueue.host = host
		msgQueue.port = port
		msgQueue.password = password
		msgQueue.user = user
		return conn, nil
	}
	err = fmt.Errorf("连接到[%s:%s %s,%s]失败,错误信息:%v\n", host, port, user, password, err)
	return nil, errorx.New(err)
}

// Send 发送消息到队列
// 参数:
//
//	 queueName:队列名称
//		msg		 :待发送的消息
//		conn	 :连接对象
//
// 返回:
//
//	error	 :错误信息
func (msgQueue *MsgQueue) send(queueName, msg string, conn *stomp.Conn) error {
	if nil == conn || msgQueue.closed {
		return fmt.Errorf("connection is nil or not connected to server")
	}
	err := conn.Send(queueName, "", []byte(msg))
	return err
}

// SendToTopic 发送主题信息
// 参数:
//
//	queue	:队列名称
//	msg		:待发送的消息
//
// 返回:
//
//	error	:错误消息
func (msgQueue *MsgQueue) SendToTopic(queue, msg string) error {
	queueName := fmt.Sprintf("%s%s", Topic, queue)
	return msgQueue.send(queueName, msg, msgQueue.connection)
}

// SendToQueue 发送消息到消息队列
// 参数:
//
//	queue	:队列名称
//	msg		:待发送的消息
//
// 返回:
//
//	error	:错误消息
func (msgQueue *MsgQueue) SendToQueue(queue, msg string) error {
	queueName := fmt.Sprintf("%s%s", Queue, queue)
	return msgQueue.send(queueName, msg, msgQueue.connection)
}

// Receive 接收队列消息,需要采用异步或者多线程的方式调用本函数,否则会阻塞
// 参数:
//
//	  sub		:订阅对象
//	  callback	:回调函数
//					:回调函数的参数
//					msg		 :消息
//					queueName:收到消息的队列名称
//
// 返回:
//
//	error		:错误信息
func (msgQueue *MsgQueue) receive(sub *stomp.Subscription, callback func(msg, queueName string)) error {
	if nil == sub {
		return errorx.New(fmt.Errorf("subscription is nil"))
	}

	for {
		v := <-sub.C
		if v != nil {
			if len(v.Body) > 0 {
				msg := string(v.Body)

				if nil != callback {
					go callback(msg, sub.Destination())
				}
			}
		} else {
			if msgQueue.closed == false {
				msgQueue.mutex.Lock()
				if msgQueue.closed == false && msgQueue.DisconnectCallback != nil {
					go msgQueue.DisconnectCallback()
					msgQueue.closed = true
				}
				msgQueue.mutex.Unlock()
			}
			break
		}
	}
	return nil
}

// SubscribeTopic 订阅主题消息
// 参数:
//
//	queues	: 队列名称
//	callback: 收到消息之后的回调
func (msgQueue *MsgQueue) SubscribeTopic(queues []string, callback func(msg, queueName string)) error {
	return msgQueue.subscribe(queues, Topic, callback)
}

// SubscribeQueue 订阅消息
// 参数:
//
//	queues	: 队列名称
//	callback: 收到消息之后的回调
func (msgQueue *MsgQueue) SubscribeQueue(queues []string, callback func(msg, queueName string)) error {
	return msgQueue.subscribe(queues, Queue, callback)
}

func (msgQueue *MsgQueue) subscribe(queues []string, queueType string, callback func(msg, queueName string)) error {
	if msgQueue.connection == nil {
		return fmt.Errorf("连接未初始化")
	}
	if queues == nil || len(queues) <= 0 {
		return fmt.Errorf("订阅的队列为空")
	}
	if msgQueue.subs == nil {
		msgQueue.subs = list.New()
	}
	for index := range queues {
		qname := queues[index]
		if qname != "" {
			q := fmt.Sprintf("%s%s", queueType, queues[index])
			sub, err := msgQueue.connection.Subscribe(q, stomp.AckAuto)
			if nil == err {
				msgQueue.subs.PushBack(sub)
				go func() {
					_ = msgQueue.receive(sub, callback)
				}()
			} else {
				return errorx.New(err)
			}
		}
	}
	return nil
}

// UnSubscriptTopic 取消主题订阅
// 参数:
//
//	queues	: 队列名称
func (msgQueue *MsgQueue) UnSubscriptTopic(queue string) error {
	return msgQueue.unSub(queue, Topic)
}

// UnSubscriptQueue 取消订阅队列
// 参数:
//
//	queues	: 队列名称
func (msgQueue *MsgQueue) UnSubscriptQueue(queue string) error {
	return msgQueue.unSub(queue, Queue)
}

func (msgQueue *MsgQueue) unSub(queue, queueType string) error {
	if msgQueue.subs == nil {
		return nil
	}
	var next *list.Element
	q := fmt.Sprintf("%s%s", queueType, queue)
	for sub := msgQueue.subs.Front(); sub != nil; sub = next {
		o, r := sub.Value.(*stomp.Subscription)
		if r && o.Destination() == q {
			msgQueue.subs.Remove(sub)
			_ = o.Unsubscribe()
			return nil
		}
		next = sub.Next()
	}
	return nil
}
